package com.xixi.canal.client.starter.monitor;

import com.xixi.canal.client.starter.common.SyncSwitch;
import com.xixi.canal.client.starter.config.CanalConfig;
import com.xixi.canal.client.starter.service.CanalClientService;
import com.xixi.canal.client.util.StringUtil;
import org.apache.commons.io.filefilter.FileFilterUtils;
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
import org.apache.commons.io.monitor.FileAlterationMonitor;
import org.apache.commons.io.monitor.FileAlterationObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.yaml.snakeyaml.Yaml;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map;

@Component
public class InstanceConfigMonitor {

    private static final Logger logger = LoggerFactory.getLogger(ApplicationConfigMonitor.class);

    private FileAlterationMonitor[] fileMonitor;

    @Resource
    private CanalConfig canalConfig;

    @Resource
    private CanalClientService canalClientService;

    @Resource
    private SyncSwitch syncSwitch;

    @PostConstruct
    public void init() {

        String destinations = canalConfig.getDestinations();
        if (!StringUtil.hasText(destinations)) return;

        String[] instances = destinations.split(",");

        fileMonitor = new FileAlterationMonitor[instances.length];
        int index = 0;
        for (String instance : instances) {
            File confDir = StringUtil.getConfDirPath(canalConfig.getConfDir(), instance);
            try {
                FileAlterationObserver observer = new FileAlterationObserver(confDir,
                        FileFilterUtils.and(FileFilterUtils.fileFileFilter(),
                                FileFilterUtils.prefixFileFilter("instance"),
                                FileFilterUtils.suffixFileFilter("yml")));
                FileListener listener = new FileListener();
                observer.addListener(listener);
                fileMonitor[index] = new FileAlterationMonitor(3000, observer);
                fileMonitor[index].start();
                index++;
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }
    @PreDestroy
    public void destroy() {
        for (FileAlterationMonitor fileAlterationMonitor : fileMonitor) {
            try {
                fileAlterationMonitor.stop();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    private class FileListener extends FileAlterationListenerAdaptor {

        @Override
        public void onFileChange(File file) {
            try {

                new Yaml().loadAs(new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8), Map.class);
                String fileName = file.getName();
                String destination = file.getParent();

                int pos = destination.lastIndexOf(File.separator);
                if (pos > -1) {
                    destination = destination.substring(pos + 1);
                }

                syncSwitch.off(destination);

                canalClientService.refresh(destination, fileName);

                syncSwitch.on(destination);
                logger.info("## Instance config reloaded.");
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

}
